﻿using System;
using System.Collections.Generic;
using System.Collections.ObjectModel;
using System.Linq;
using System.Text;
using System.IO;
using System.Threading;
using System.Data;
using System.Data.SqlClient;
using System.Configuration;
using System.Globalization;

namespace Laharsub.Server.Wcf
{
    public class SqlPubSubBackend : IPubSubBackend
    {
        static string connectionString;

        static string ConnectionString
        {
            get
            {
                if (connectionString == null)
                {
                    ConnectionStringSettings s = ConfigurationManager.ConnectionStrings["SqlPubSubBacked"];
                    if (s == null)
                    {
                        throw new InvalidOperationException("A connection string 'SqlPubSubBacked' must be specified in the configuration file.");
                    }
                    connectionString = s.ConnectionString;
                }
                return connectionString;
            }
        }

        public SqlPubSubBackend()
        {
            new SqlClientPermission(System.Security.Permissions.PermissionState.Unrestricted).Demand();
            if (!SqlDependency.Start(ConnectionString))
            {
                throw new InvalidOperationException("SqlDependency listener cannot be started.");
            }
        }

        static SqlConnection GetConnection()
        {
            SqlConnection result = new SqlConnection(ConnectionString);
            result.Open();
            return result;
        }

        public IAsyncResult BeginCreateTopic(AsyncCallback callback, object state)
        {
            SqlCommand command = new SqlCommand("createTopic", GetConnection()) { CommandType = CommandType.StoredProcedure };
            return new ScalarAsyncResult<int>(
                delegate(AsyncCallback c, object s) // nested begin handler
                {
                    return command.BeginExecuteReader(c, s, CommandBehavior.CloseConnection);
                },
                delegate(IAsyncResult r)   // nested end handler
                {
                    using (SqlDataReader reader = command.EndExecuteReader(r))
                    {
                        if (reader.Read())
                        {
                            int topicId = (int)reader.GetDecimal(0);
                            return topicId;
                        }
                        else
                        {
                            throw new InvalidOperationException("Unable to create a new topic.");
                        }
                    }
                },
                callback,
                state
            );
        }

        public int EndCreateTopic(IAsyncResult result)
        {
            return ScalarAsyncResult<int>.End(result);
        }

        public IAsyncResult BeginPublishMessage(int topicId, string contentType, Stream body, AsyncCallback callback, object state)
        {
            return new SqlPublishMessageAsyncResult(this, topicId, contentType, body, callback, state);
        }

        public int EndPublishMessage(IAsyncResult result)
        {
            return SqlPublishMessageAsyncResult.End(result);
        }

        public IAsyncResult BeginSubscribe(IDictionary<int, int> subscriptions, TimeSpan timeout, AsyncCallback callback, object state)
        {
            return new VolatileSubscriptionAsyncResult(this, subscriptions, timeout, callback, state);
        }

        public IEnumerable<PubSubMessage> EndSubscribe(IAsyncResult result)
        {
            return VolatileSubscriptionAsyncResult.End(result);
        }

        class VolatileSubscriptionAsyncResult : IAsyncResult
        {
            static TimerCallback resumeOnTimeout = new TimerCallback(ResumeOnTimeout);
            static AsyncCallback onGetMessageIdCompleted = new AsyncCallback(OnGetMaxMessageIdCompleted);
            static AsyncCallback onQueryCompleted = new AsyncCallback(OnQueryCompleted);
            static AsyncCallback onNotificationQueryCompleted = new AsyncCallback(OnNotificationQueryCompleted);

            SqlPubSubBackend parent;
            AsyncCallback callback;
            object state;
            Exception exception;
            bool completedSynchronously;
            bool completed;
            ManualResetEvent waitHandle = new ManualResetEvent(false);
            Timer timer;
            PubSubMessage[] messages;
            SqlDependency sqlDependency;
            string subscriptionQuery;
            TimeSpan timeout;
            int maxMessageId;
            IDictionary<int, int> subscriptions;
            OnChangeEventHandler onSqlDependencyChanged;
            bool secondQuery;
            bool timedout;
            object syncRoot = new object();

            public VolatileSubscriptionAsyncResult(SqlPubSubBackend parent, IDictionary<int, int> subscriptions, TimeSpan timeout, AsyncCallback callback, object state)
            {
                this.parent = parent;
                this.callback = callback;
                this.state = state;
                this.timeout = timeout;
                this.subscriptions = subscriptions;
                this.onSqlDependencyChanged = new OnChangeEventHandler(this.OnSqlDependencyChanged);
                this.timer = new Timer(resumeOnTimeout, this, timeout, TimeSpan.Zero);
                this.StartInitializeSubscriptionQuery();
            }

            void StartInitializeSubscriptionQuery()
            {
                if (this.subscriptions.Values.Any(x => x <= 0))
                {
                    SqlCommand command = new SqlCommand("getMaxMessageId", GetConnection()) { CommandType = CommandType.StoredProcedure };
                    IAsyncResult result = new ScalarAsyncResult<int>(
                        delegate(AsyncCallback c, object s) // nested begin handler
                        {
                            return command.BeginExecuteReader(c, s, CommandBehavior.CloseConnection);
                        },
                        delegate(IAsyncResult r)   // nested end handler
                        {
                            using (SqlDataReader reader = command.EndExecuteReader(r))
                            {
                                return (reader.Read())
                                    ? (int)reader.GetInt64(0)
                                    : 0;
                            }
                        },
                        onGetMessageIdCompleted,
                        this
                    );
                    if (result.CompletedSynchronously)
                    {
                        this.GetMaxMessageIdCompleted(result, true);
                    }
                }
                else
                {
                    this.StartFirstQuery(true);
                }
            }

            static void OnGetMaxMessageIdCompleted(IAsyncResult result)
            {
                if (!result.CompletedSynchronously)
                {
                    ((VolatileSubscriptionAsyncResult)result.AsyncState).GetMaxMessageIdCompleted(result, false);
                }
            }

            void GetMaxMessageIdCompleted(IAsyncResult result, bool synchronous)
            {
                try
                {
                    this.maxMessageId = ScalarAsyncResult<int>.End(result) + 1;
                    this.StartFirstQuery(synchronous);
                }
                catch (Exception e)
                {
                    if (synchronous)
                    {
                        throw;
                    }
                    this.Complete(synchronous, e);
                }
            }

            void StartFirstQuery(bool synchronous)
            {
                this.InitializeSubscriptionQueryCore();
                this.CreateSqlDependencies();
                this.StartQuery(synchronous);
            }

            void InitializeSubscriptionQueryCore()
            {
                StringBuilder sb = null;
                foreach (KeyValuePair<int, int> s in this.subscriptions)
                {
                    if (sb == null)
                    {
                        sb = new StringBuilder("select topicid, messageid, contenttype, body from messages where");
                    }
                    else
                    {
                        sb.Append(" or");
                    }
                    sb.Append(" topicid=" + s.Key.ToString(CultureInfo.InvariantCulture));
                    sb.Append(" and messageid>=" +
                        (s.Value > 0
                        ? s.Value.ToString(CultureInfo.InvariantCulture)
                        : this.maxMessageId.ToString(CultureInfo.InvariantCulture)));
                }
                this.subscriptionQuery = sb.ToString();
            }

            void StartQuery(bool synchronous)
            {
                SqlCommand command = new SqlCommand(this.subscriptionQuery, GetConnection());
                IAsyncResult result = new ScalarAsyncResult<PubSubMessage[]>(
                    delegate(AsyncCallback c, object s) // nested begin handler
                    {
                        return command.BeginExecuteReader(c, s, CommandBehavior.CloseConnection);
                    },
                    delegate(IAsyncResult r) // nested end handler
                    {
                        List<PubSubMessage> messages = new List<PubSubMessage>();
                        using (SqlDataReader reader = command.EndExecuteReader(r))
                        {
                            while (reader.Read())
                            {
                                messages.Add(new PubSubMessage
                                {
                                    TopicId = (int)reader.GetInt64(0),
                                    MessageId = (int)reader.GetInt64(1),
                                    ContentType = reader.GetString(2),
                                    Body = Encoding.UTF8.GetBytes(reader.GetString(3))
                                });
                            }
                        }
                        return messages.Count > 0 ? messages.ToArray() : null;
                    },
                    onQueryCompleted,
                    this);
                if (result.CompletedSynchronously)
                {
                    this.QueryCompleted(result, synchronous);
                }
            }

            static void OnQueryCompleted(IAsyncResult result)
            {
                if (!result.CompletedSynchronously)
                {
                    ((VolatileSubscriptionAsyncResult)result.AsyncState).QueryCompleted(result, false);
                }
            }

            void QueryCompleted(IAsyncResult result, bool synchronous)
            {
                bool completeResult = false;
                try
                {
                    this.messages = ScalarAsyncResult<PubSubMessage[]>.End(result);
                    if (this.messages != null || secondQuery)
                    {
                        completeResult = true;
                    }
                }
                catch (Exception e)
                {
                    if (synchronous)
                    {
                        throw;
                    }
                    this.Complete(synchronous, e);
                }
                if (completeResult)
                {
                    this.Complete(synchronous, null);
                }
            }

            void CreateSqlDependencies()
            {
                this.sqlDependency = new SqlDependency(null, null, (int)this.timeout.TotalSeconds);
                this.sqlDependency.OnChange += this.onSqlDependencyChanged;
                foreach (int key in subscriptions.Keys)
                {
                    SqlCommand command = new SqlCommand(
                        "select dbo.messages.messageid from dbo.messages where dbo.messages.topicid=" + key.ToString(CultureInfo.InvariantCulture) + ";",
                        GetConnection());
                    this.sqlDependency.AddCommandDependency(command);
                    command.BeginExecuteReader(onNotificationQueryCompleted, command, CommandBehavior.CloseConnection);
                }
            }

            static void OnNotificationQueryCompleted(IAsyncResult result)
            {
                try
                {
                    using (((SqlCommand)result.AsyncState).EndExecuteReader(result)) { } // empty
                }
                catch (Exception)
                {
                }
            }

            void OnSqlDependencyChanged(object sender, SqlNotificationEventArgs e)
            {
                SqlDependency dependency = (SqlDependency)sender;
                dependency.OnChange -= this.onSqlDependencyChanged;
                bool startSecondQuery = false;
                bool timeout = false;
                if (!this.timedout && !this.secondQuery)
                {
                    lock (this.syncRoot)
                    {
                        if (!this.timedout && !this.secondQuery)
                        {
                            if (e.Info == SqlNotificationInfo.Insert)
                            {
                                startSecondQuery = this.secondQuery = true;
                            }
                            else
                            {
                                timeout = this.timedout = true;
                            }
                        }
                    }
                    if (startSecondQuery)
                    {
                        if (!this.completed)
                        {
                            try
                            {
                                this.StartQuery(false);
                            }
                            catch (Exception ex)
                            {
                                if (!this.completed)
                                {
                                    lock (this.syncRoot)
                                    {
                                        if (!this.completed)
                                        {
                                            this.Complete(false, ex);
                                        }
                                    }
                                }
                            }
                        }
                    }
                    else if (timeout && !this.completed)
                    {
                        lock (this.syncRoot)
                        {
                            if (!this.completed)
                            {
                                this.Complete(false, null);
                            }
                        }
                    }
                }
            }

            public static IEnumerable<PubSubMessage> End(IAsyncResult result)
            {
                VolatileSubscriptionAsyncResult r = result as VolatileSubscriptionAsyncResult;
                if (r == null)
                {
                    throw new InvalidOperationException("Invalid async result.");
                }
                r.waitHandle.WaitOne();
                r.CancelSqlDependencies();
                if (r.exception != null)
                {
                    throw r.exception;
                }
                else
                {
                    return r.messages;
                }
            }

            static void ResumeOnTimeout(object state)
            {
                ((VolatileSubscriptionAsyncResult)state).ResumeOnTimeout();
            }

            void ResumeOnTimeout()
            {
                if (!this.completed && !this.timedout && !this.secondQuery)
                {
                    lock (this.syncRoot)
                    {
                        if (!this.completed && !this.timedout && !this.secondQuery)
                        {
                            this.Complete(false, null);
                        }
                    }
                }
                this.CancelSqlDependencies();
            }

            void CancelSqlDependencies()
            {
                this.sqlDependency.OnChange -= this.onSqlDependencyChanged;
            }

            void Complete(bool completedSynchronously, Exception e)
            {
                if (!this.completed)
                {
                    lock (this.syncRoot)
                    {
                        if (!this.completed)
                        {
                            this.exception = e;
                            this.completedSynchronously = completedSynchronously;
                            this.completed = true;
                            this.waitHandle.Set();
                            if (this.callback != null)
                            {
                                this.callback(this);
                            }
                        }
                    }
                }
            }

            public object AsyncState
            {
                get { return this.state; }
            }

            public WaitHandle AsyncWaitHandle
            {
                get { return this.waitHandle; }
            }

            public bool CompletedSynchronously
            {
                get { return this.completedSynchronously; }
            }

            public bool IsCompleted
            {
                get { return this.completed; }
            }
        }

        class SqlPublishMessageAsyncResult : PublishMessageAsyncResult
        {
            static AsyncCallback onPublishCompleted = new AsyncCallback(OnPublishCompleted);

            SqlPubSubBackend parent;

            public SqlPublishMessageAsyncResult(SqlPubSubBackend parent, int topicId, string contentType, Stream body, AsyncCallback callback, object state)
                : base(topicId, contentType, body, callback, state)
            {
                this.parent = parent;
                this.Start(); // asynchronously buffers the message body and calls StartPublish
            }

            public override void StartPublish(bool synchronous)
            {
                try
                {
                    SqlCommand command = new SqlCommand("createMessage", GetConnection()) { CommandType = CommandType.StoredProcedure };
                    command.Parameters.Add(new SqlParameter("@topicid", this.Message.TopicId));
                    command.Parameters.Add(new SqlParameter("@contenttype", this.Message.ContentType));
                    command.Parameters.Add(new SqlParameter("@body", this.Message.GetBodyAsString()));
                    IAsyncResult result = new ScalarAsyncResult<int>(
                        delegate(AsyncCallback c, object s) // nested begin handler
                        {
                            return command.BeginExecuteReader(c, s, CommandBehavior.CloseConnection);
                        },
                        delegate(IAsyncResult r)   // nested end handler
                        {
                            using (SqlDataReader reader = command.EndExecuteReader(r))
                            {
                                if (reader.Read())
                                {
                                    int topicId = (int)reader.GetDecimal(0);
                                    return topicId;
                                }
                                else
                                {
                                    throw new InvalidOperationException("Unable to publish message.");
                                }
                            }
                        },
                        onPublishCompleted,
                        this
                    );
                    if (result.CompletedSynchronously)
                    {
                        this.PublishCompleted(result, true);
                    }
                }
                catch (Exception e)
                {
                    this.Complete(synchronous, e);
                }
            }

            static void OnPublishCompleted(IAsyncResult result)
            {
                if (!result.CompletedSynchronously)
                {
                    ((SqlPublishMessageAsyncResult)result.AsyncState).PublishCompleted(result, false);
                }
            }

            void PublishCompleted(IAsyncResult result, bool synchronous)
            {
                try
                {
                    this.Message.MessageId = ScalarAsyncResult<int>.End(result);
                }
                catch (Exception e)
                {
                    this.Complete(synchronous, e);
                }
                this.Complete(synchronous, null);
            }
        }
    }
}
